package com.chukun.flink.stream.state.key;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author chukun
 * @version 1.0.0
 * @description 带状态的flatmap操作符 valueState的基本使用
 * @createTime 2022年05月15日 22:59:00
 */
public class ValueStateFlatMap extends RichFlatMapFunction<Tuple2<Integer, Integer>, Tuple2<Integer,Integer>> {

    private static final Logger logger = LoggerFactory.getLogger(ValueStateFlatMap.class);

    // 创建ValueState类型的状态结构
    private transient ValueState<Tuple2<Integer,Integer>> valueState;

    @Override
    public void flatMap(Tuple2<Integer, Integer> input, Collector<Tuple2<Integer, Integer>> collector) throws Exception {
        // 获取存储在valueState状态中的数据
        Tuple2<Integer, Integer> currentState = valueState.value();
        logger.info("{} currentState before : {} ,input : {}",Thread.currentThread().getName(), currentState, input);
        System.out.println("currentState before : " + currentState+", input : "+input);
        if (currentState == null) {
            // 更新valueState数据
            currentState = input;
        } else {
            currentState.f1 = currentState.f1 + input.f1;
        }
        if (currentState.f1 % 10 > 6) {
            // 若状态中数据的f1字段除以10的余数大于6，则将该状态中的数据发送给下游操作符
            collector.collect(currentState);
            // 清除状态标记
            valueState.clear();
        } else {
            // 更新valueState
            valueState.update(currentState);
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        logger.info("{},{}", Thread.currentThread().getName(), "恢复或初始化状态");

        // 创建一个ValueStateDescriptor状态描述符，指定名称为valueStateFlatMap，类型为 Tuple2<Integer, Integer>
        ValueStateDescriptor<Tuple2<Integer,Integer>> descriptor = new ValueStateDescriptor<>(
                "ValueStateFlatMap",
                TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {}));
        // 初始化valueState
        valueState = getRuntimeContext().getState(descriptor);
        logger.info("初始化 valueState ： {}", valueState);
    }

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 封装了数据源函数，以及keyBy操作的数据流
        KeyedStream<Tuple2<Integer, Integer>, Integer> keyedStream = KeyStateBase.before(env);

        // 在keyedStream中使用有状态的FlatMap操作符
        DataStream<Tuple2<Integer, Integer>> resultStream = keyedStream.flatMap(new ValueStateFlatMap());

        resultStream.print("输出结果 : ");

        env.execute("ValueStateFlatMap");
    }
}
